import kafka from "node-rdkafka"
import { conf } from "../confParse.js"
import { eventHandler, EventCode} from "../event/eventHandler.js"
import {FetcherFactor} from '../Fetchers/fetcherFactory.js'
import { iArg } from "../iArg.js"

async function reqHandle(reqStream:kafka.ConsumerStream,respProducer:kafka.Producer,conf:conf){
    reqStream.on("data", async (data: kafka.Message) => {
        eventHandler.handle(EventCode.NORMAL, "get data " + data.value?.toString())
        if(data.value){
            let arg:iArg=JSON.parse(data.value.toString())
            let fetcher=FetcherFactor.create(arg.requestType)
            if(fetcher!==undefined){
                let respData=await fetcher.fetch(arg)
                eventHandler.handle(EventCode.PAESE_REQ_ERROR,respData,new Error())
                respProducer.produce(conf.respTopic,null,Buffer.from(respData),data.key)
            }else{
                eventHandler.handle(EventCode.PAESE_REQ_ERROR,data.value.toString(),new Error())
            }
            
        }
        
    })
}

export {reqHandle}